package cn.lsh.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class HbaseUtil {

	private HbaseUtil() {
	}

	public static String getCellValue(Result result, String cf, String col) {
		return Bytes.toString(result.getValue(Bytes.toBytes(cf), Bytes.toBytes(col)));
	}

	/**
	 * 转换Result为Map
	 *
	 * @param result Result
	 * @param t value类型
	 * @return Map 映射为 columnFamily：column：value
	 */
	public static <T> Map<String, Map<String, T>> resultToMap(Result result, Class<T> t) {
		if (result.advance()) {
			Map<String, Map<String, T>> rowKeyCell = new HashMap<>();
			List<Cell> cells = result.listCells();
			for (Cell cell : cells) {
				Map<String, T> columnValueMap;
				String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
				if (rowKeyCell.containsKey(columnFamily)) {
					columnValueMap = rowKeyCell.get(columnFamily);
					columnValueMap.put(Bytes.toString(CellUtil.cloneQualifier(cell)), getValuFormCell(cell, t));
				} else {
					columnValueMap = new HashMap<>();
					columnValueMap.put(Bytes.toString(CellUtil.cloneQualifier(cell)), getValuFormCell(cell, t));
					rowKeyCell.put(columnFamily, columnValueMap);
				}
			}
			return rowKeyCell;
		}
		return null;
	}

	public static <T> T getValuFormCell(Cell cell, Class<T> t) {
		byte[] bytes = CellUtil.cloneValue(cell);
		String tName = t.getSimpleName();
		if (String.class.getSimpleName().equals(tName)) {
			return (T) Bytes.toString(bytes);
		} else if (Integer.class.getSimpleName().equals(tName)) {
			return (T) Integer.valueOf(Bytes.toInt(bytes));
		} else if (Long.class.getSimpleName().equals(tName)) {
			return (T) Long.valueOf(Bytes.toLong(bytes));
		} else if (Double.class.getSimpleName().equals(tName)) {
			return (T) Double.valueOf(Bytes.toDouble(bytes));
		} else if (Short.class.getSimpleName().equals(tName)) {
			return (T) Short.valueOf(Bytes.toShort(bytes));
		} else if (Float.class.getSimpleName().equals(tName)) {
			return (T) Float.valueOf(Bytes.toFloat(bytes));
		}
		return null;
	}

	/**
	 * 转换ResultScanner为Map映射，rowKey：columnFamily：columns
	 *
	 * @param rs ResultScanner数据扫描器
	 * @return Map 映射类型 rowKey：columnFamily：columns
	 */
	public static Map<String, Map<String, List<String>>> resultScannerToMap(ResultScanner rs) {
		Map<String, Map<String, List<String>>> rowKeyColumnMap = new HashMap<>();
		for (Result result : rs) {
			Map<String, List<String>> columnMap;
			String rowKey = Bytes.toString(result.getRow());
			boolean exists = rowKeyColumnMap.containsKey(rowKey);
			if (exists) {
				columnMap = rowKeyColumnMap.get(rowKey);
			} else {
				columnMap = new HashMap<>();
			}
			if (result.advance()) {
				for (Cell cell : result.listCells()) {
					String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
					List<String> columns;
					if (columnMap.containsKey(columnFamily)) {
						columns = columnMap.get(columnFamily);
						columns.add(Bytes.toString(CellUtil.cloneQualifier(cell)));
					} else {
						columns = new ArrayList<>();
						columns.add(Bytes.toString(CellUtil.cloneQualifier(cell)));
						columnMap.put(columnFamily, columns);
					}
				}
			}
			if (!exists) {
				rowKeyColumnMap.put(rowKey, columnMap);
			}
		}
		//ResultScanner实时RPC读取，必须在读取完数据后关闭，重复关闭不会报错
		rs.close();
		return rowKeyColumnMap;
	}

}
